package net.mingsoft.report_data.biz.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import lombok.extern.slf4j.Slf4j;
import net.mingsoft.basic.util.DateUtil;
import net.mingsoft.report_data.biz.ZkDataMessageBiz;
import net.mingsoft.report_data.constant.ZkDataConstant;
import net.mingsoft.report_data.dao.ZkMessageDao;
import net.mingsoft.report_data.entity.ZkDataMessageEntity;
import net.mingsoft.report_data.entity.ZkDataMessageLogEntity;
import net.mingsoft.report_data.kafka.avro.ZkReport;
import net.mingsoft.report_data.kafka.entity.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author gengfa
 * @date 22/03/28
 */
@Service
@Slf4j
public class ZkDataMessageBizImpl implements ZkDataMessageBiz {

    @Resource
    ZkMessageDao messageDao;

    private static ZkMessageDao zkMessageDao;

    private static final int MAX_SIZE_PER_TIME = 500;

    @PostConstruct
    public void init() {
        ZkDataMessageBizImpl.zkMessageDao = this.messageDao;
    }



    private static final Map<String, Function<Object, Boolean>> SAVE_FUNC_MAP = new HashMap<>(16);

    static {
        SAVE_FUNC_MAP.put("UserCountByConstantly", obj -> {
            if(Optional.ofNullable(obj).isPresent()) {
                final UserCountByConstantlyEntity e = (UserCountByConstantlyEntity) obj;
                return saveUserDto(e.getProvinceUser(), 1, e.getTime())
                        && saveUserDto(e.getOutsideUser(), 2, e.getTime())
                        && saveUserDto(e.getAbroadUser(), 3, e.getTime());
            }
            return false;
        });

        SAVE_FUNC_MAP.put("UserHeatByConstantly", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final UserHeatByConstantlyEntity e = (UserHeatByConstantlyEntity) obj;
                if (e.getHeatChart().size() <= 0) {
                    log.warn("empty data");
                    return true;
                }
                return e.getHeatChart().size() == zkMessageDao.insertHainanPopulation(e.getHeatChart(), e.getTime());
            }
            return false;
        });

        SAVE_FUNC_MAP.put("TravelUserByDay", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final TravelUserByDayEntity e = (TravelUserByDayEntity) obj;
                return e.getUserAreaList().size() == zkMessageDao.insertTravelUserPopulation(e.getUserAreaList(), e.getTime());
            }
            return false;
        });

        SAVE_FUNC_MAP.put("CoastalUserByDay", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final CoastalUserByDayEntity e = (CoastalUserByDayEntity) obj;
                if (e.getUserAreaList().size() > 0) {
                    return e.getUserAreaList().size() == zkMessageDao.insertCoastalUserPopulation(e.getUserAreaList(), e.getTime());
                } else {
                    log.warn("empty data");
                    return true;
                }
            }
            return false;
        });

        SAVE_FUNC_MAP.put("SmuggleUserByDay", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final SmuggleUserByDayEntity e = (SmuggleUserByDayEntity) obj;
                if (e.getTravelAreaList().size() <= 0 || e.getTravelFrequencyList().size() <= 0) {
                    log.warn("empty data");
                    return true;
                }
                return e.getTravelAreaList().size() == zkMessageDao.insertSmuggleUserAreaPopulation(e.getTravelAreaList(), e.getTime())
                        && e.getTravelFrequencyList().size() == zkMessageDao.insertSmugglerTravelUserCountPopulation(e.getTravelFrequencyList(), e.getTime())
                        && 1 == zkMessageDao.insertSmugglerTravelTimePopulation(e.getTravelTimeDto(), e.getTime())
                        && 1 == zkMessageDao.insertSmugglerTravelDistanceDto(e.getTravelDistanceDto(), e.getTime());
            }
            return false;
        });

        SAVE_FUNC_MAP.put("SmuggleUserWordLiveByDay", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final SmuggleUserWordLiveByDayEntity e = (SmuggleUserWordLiveByDayEntity) obj;
                if (e.getLivePlaceDto().size() <= 0 || e.getWorkPlaceDto().size() <= 0) {
                    log.warn("empty data");
                    return true;
                }
                return e.getLivePlaceDto().size() == zkMessageDao.insertSmugglerHomePopulation(e.getLivePlaceDto(), e.getTime())
                        && e.getWorkPlaceDto().size() == zkMessageDao.insertSmugglerWorkPopulation(e.getWorkPlaceDto(), e.getTime());
            }
            return false;
        });
        SAVE_FUNC_MAP.put("PermanentPopulation", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final PermanentPopulationEntity e = (PermanentPopulationEntity) obj;
                return e.getData().size() == zkMessageDao.insertPermanentPopulation(e.getData(), e.getTime());
            }
            return false;
        });
        SAVE_FUNC_MAP.put("FlowPopulationPerHour", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final FlowPopulationPerHourEntity e = (FlowPopulationPerHourEntity) obj;
                return e.getData().size() == zkMessageDao.insertFlowPopulation(e.getData(), e.getTime());
            }
            return false;
        });
        SAVE_FUNC_MAP.put("CoastalPermanentPopulation", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final CoastalPermanentPopulationEntity e = (CoastalPermanentPopulationEntity) obj;
                return e.getData().size() == zkMessageDao.insertCoastalPermanentPopulation(e.getData(), e.getTime());
            }
            return false;
        });
        SAVE_FUNC_MAP.put("CoastalWorkHomePopulation", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final CoastalWorkHomePopulationEntity e = (CoastalWorkHomePopulationEntity) obj;
                if (e.getWorkPlaceDto().size() <= 0 || e.getLivePlaceDto().size() <= 0) {
                    log.warn("empty data");
                    return true;
                }
                return e.getLivePlaceDto().size() == zkMessageDao.insertCoastalHomePopulation(e.getLivePlaceDto(), e.getTime())
                        && e.getWorkPlaceDto().size() == zkMessageDao.insertCoastalWorkPopulation(e.getWorkPlaceDto(), e.getTime());
            }
            return false;
        });
        SAVE_FUNC_MAP.put("CityPopulationFlowPerMonth", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final CityPopulationFlowPerMonthEntity e = (CityPopulationFlowPerMonthEntity) obj;
                String formatTime = DateUtil.transferDateFormat(e.getTime(), DateUtil.DATE_FORMAT_YEAR_MONTH, DateUtil.DATETIME_FORMAT_STR);
                return e.getData().size() == zkMessageDao.insertCityFlowPopulation(e.getData(), formatTime);
            }
            return true;
        });
        SAVE_FUNC_MAP.put("CitiesBetweenPopulationFlow", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final CitiesBetweenPopulationFlowEntity e = (CitiesBetweenPopulationFlowEntity) obj;
                String formatTime = DateUtil.transferDateFormat(e.getTime(), DateUtil.DATE_FORMAT_YEAR_MONTH, DateUtil.DATETIME_FORMAT_STR);
                return e.getData().size() == zkMessageDao.insertCityMigrationPopulation(e.getData(), formatTime);
            }
            return false;
        });
        // TODO 需要提供基站列表
        SAVE_FUNC_MAP.put("SpecificEmployeeMonitor", obj -> true);
        SAVE_FUNC_MAP.put("ODPopulationPerHour", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final ODPopulationPerHourEntity e = (ODPopulationPerHourEntity) obj;
                return 1 == zkMessageDao.insertODPerHourPopulation(e);
            }
            return false;
        });
        SAVE_FUNC_MAP.put("ODDistanceAndCostTimePerDay", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final ODDistanceAndCostTimePerDayEntity e = (ODDistanceAndCostTimePerDayEntity) obj;
                if (e.getAvgData().size() <= 0 || e.getCostTimeData().size() <= 0 || e.getDistanceData().size() <= 0) {
                    log.warn("empty data");
                    return true;
                }
                return e.getAvgData().size() == zkMessageDao.insertODAvgData(e.getAvgData(), e.getTime())
                        && e.getCostTimeData().size() == zkMessageDao.insertODCostTimeData(e.getCostTimeData(), e.getTime())
                        && e.getDistanceData().size() == zkMessageDao.insertODDistanceData(e.getDistanceData(), e.getTime());
            }
            return false;
        });
        SAVE_FUNC_MAP.put("ODCostTimeDistributionPerHour", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final ODCostTimeDistributionPerHourEntity e = (ODCostTimeDistributionPerHourEntity) obj;
                if (e.getData().size() > 0) {
                    return e.getData().size() == zkMessageDao.insertODCostTimeDistribution(e.getData(), e.getTime());
                } else {
                    log.warn("empty data");
                    return true;
                }
            }
            return false;
        });
        SAVE_FUNC_MAP.put("ODWorkHomeDistributionPerMonth", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final ODWorkHomeDistributionPerMonthEntity e = (ODWorkHomeDistributionPerMonthEntity) obj;
                final String formatDate = DateUtil.transferDateFormat(e.getTime(), DateUtil.DATE_FORMAT_YEAR__MONTH, DateUtil.DATETIME_FORMAT_STR);
                return e.getData().size() == zkMessageDao.insertODWorkHomeDistributionData(e.getData(), formatDate);
            }
            return false;
        });
        SAVE_FUNC_MAP.put("TopHotAreaPerDay", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final TopHotAreaPerDayEntity e = (TopHotAreaPerDayEntity) obj;
                if (e.getData().size() <= 0) {
                    log.warn("empty data");
                    return true;
                }
                for (TopHotAreaDto datum : e.getData()) {
                    datum.setDay(DateUtil.transferDateFormat(datum.getDay(), DateUtil.DATE_FORMAT_STR_L, DateUtil.DATE_FORMAT_STR));
                }
                return e.getData().size() == zkMessageDao.insertTopHotAreaPerDayData(e.getData(), e.getTime());
            }
           return false;
        });
        SAVE_FUNC_MAP.put("InIslandFlowPopulationPerDay", obj -> {
            if (Optional.ofNullable(obj).isPresent()) {
                final InIslandFlowPopulationPerDayEntity e = (InIslandFlowPopulationPerDayEntity) obj;
                return e.getData().size() == zkMessageDao.insertInIslandFlowPopulationData(e.getData(), e.getTime());
            }
            return false;
        });

    }

    @Override
    public boolean saveMessage(ZkReport message) {
        return 1 == messageDao.insertOneMessage(message);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void parseMessage() {
        log.info("parseMessage() start");
        int total = 0;
        List<ZkDataMessageEntity> recentlyMessage = getRecentlyMessage();
        while (recentlyMessage.size() > 0 && total <= MAX_SIZE_PER_TIME) {
            total += recentlyMessage.size();
            Boolean status = false;
            for (ZkDataMessageEntity it : recentlyMessage) {
                try {
                    // TODO parse message by it.getUri()
                    final String key = getUriSubString(it.getUri());
                    final String className = getClassName(key);
                    final Object object = JSON.parseObject(it.getData(), getVoClass(className));
                    status = SAVE_FUNC_MAP.get(key).apply(object);
                    if (!status) {
                        log.error("[ROLLBACK] [{}] parse failed, id: {}, message_id: {}", it.getUri(), it.getId(), it.getMessageId());
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                    }
                } catch (JSONException e) {
                    log.error("[JSONException] [{}] parse failed, id: {}, message_id: {} | cause: {}", it.getUri(), it.getId(), it.getMessageId(), e.getMessage());
                } catch (Exception e) {
                    log.error("[ROLLBACK] [{}] parse failed, id: {}, message_id: {} | cause: {}", it.getUri(), it.getId(), it.getMessageId(), e.getMessage());
                    TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                }
            }
            // 保存日志;
            saveParseLog(recentlyMessage, status);
            log.info("[progress] {} message saved", total);
            // next list
            recentlyMessage = getRecentlyMessage();
        }
        log.info("parseMessage() finished: {} messages", total);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void parseMessage(List<String> types) {
        log.info("parseMessage() start");
        int total = 0;
        List<ZkDataMessageEntity> recentlyMessage = messageDao.selectMessageByTypes(types);
        while (recentlyMessage.size() > 0 && total <= MAX_SIZE_PER_TIME) {
            total += recentlyMessage.size();
            Boolean status = false;
            for (ZkDataMessageEntity it : recentlyMessage) {
                // TODO parse message by it.getUri()
                final String key = getUriSubString(it.getUri());
                final String className = getClassName(key);
                final Object object = JSON.parseObject(it.getData(), getVoClass(className));
                status = SAVE_FUNC_MAP.get(key).apply(object);
                if (!status) {
                    TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                }
            }
            // 保存日志;
            //saveParseLog(recentlyMessage, status);
            log.info("[progress] {} message saved", total);
            // next list
            //recentlyMessage = getRecentlyMessage();
        }
        log.info("parseMessage() finished: {} messages", total);
    }

    /**
     * 保存消息解析日志
     * @param entities 消息列表
     */
    void saveParseLog(List<ZkDataMessageEntity> entities, boolean status) {
        ZkDataMessageLogEntity logEntity = new ZkDataMessageLogEntity();
        logEntity.setCount(entities.size());
        if (entities.size() > 0) {
            logEntity.setStartId(entities.get(0).getId())
                    .setEndId(entities.get(entities.size()-1).getId())
                    .setIds(entities.stream().map(o -> String.valueOf(o.getId())).collect(
                            Collectors.joining(",")));
        }
        logEntity.setStatus(status?ZkDataConstant.MESSAGE_PARSED_SUCCESS:ZkDataConstant.MESSAGE_PARSED_FAILED);
        messageDao.saveMessageParseLog(logEntity, DateUtil.now6());
        log.info("[progress] message parsing logs saved");
    }

    /**
     * 获取未解析的消息
     * @return kafka消息
     */
    private List<ZkDataMessageEntity> getRecentlyMessage() {
        log.info("parseMessage() getRecentlyMessage()");
        // 最后解析日志
        ZkDataMessageLogEntity entity = messageDao.selectLatestMessageLog();
        long endMessageId = 0;
        if (entity != null) {
            endMessageId = entity.getEndId();
        }
        // 查询未解析的
        return messageDao.selectMessagesBigThanId(endMessageId);
    }

    /**
     * 获取key
     * @param uri uri
     * @return key for map
     */
    private String getUriSubString(String uri) {
        return uri.substring(ZkDataConstant.URI_PREFIX.length());
    }

    /**
     * 获取完成包类名
     * @param s s
     * @return 返回类完整包名
     */
    private String getClassName(String s) {
        return ZkDataConstant.MESSAGE_PACKAGE_NAME + s + ZkDataConstant.ENTITY_SUFFIX;
    }

    /**
     * 获取消息JSON对应的实体类
     * @param className className
     * @return 类
     */
    private Class<?> getVoClass(String className) {
        try {
            return Class.forName(className);
        } catch (ClassNotFoundException e) {
            log.error("not found {}, {}", className, e.getMessage());
        }
        return String.class;
    }

    /**
     * TODO 保存urn:ga-send:UserCountByConstantly数据
     * @param userDto userDto
     * @param target 目标表
     */
    private static boolean saveUserDto(UserDto userDto, int target, String time) {
        if (Optional.ofNullable(userDto).isPresent()) {
            List<UserAreaDto> userAreaList = userDto.getUserAreaList();
            if (userAreaList.size() <= 0) {
                log.warn("empty data");
                return true;
            }
            int rows = 0;
            if (target == 1) {
                rows = zkMessageDao.insertActivePopulation(userAreaList, 1, time);
            } else if (target == 2) {
                rows = zkMessageDao.insertActivePopulation(userAreaList, 0, time);
            } else if (target == 3) {
                rows = zkMessageDao.insertForeignPopulation(userAreaList, time);
            }
            return rows == userAreaList.size();
        }
        return true;
    }
}
